package studio.raptor.dispatcher.server.dao;

import com.google.common.base.Joiner;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ListMultimap;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import javax.sql.DataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Repository;
import studio.raptor.databus.command.EventMetaData;
import studio.raptor.dispatcher.common.constant.EventParamConst;
import studio.raptor.dispatcher.common.constant.EventStatus;
import studio.raptor.dispatcher.server.dto.EventMetaDataWrapper;
import studio.raptor.dispatcher.server.dto.SendResponse;
import studio.raptor.dispatcher.server.entity.EventTrace;
import studio.raptor.hub.common.util.JdbcUtils;
import studio.raptor.hubout.sdk.sender.Response;

/**
 * 保存事件发送记录dao类
 *
 * @author jack
 * @since 1.0
 */
@Repository
public class EventDao {

  private static final Logger log = LoggerFactory.getLogger(EventDao.class);

  static String INSERT_EVENT_TRACE =
      "INSERT INTO event_trace (trace_id,object_key,code,status,kind_id,event_type,retry_times,create_date,last_update_date,resp_code)"
          + " VALUES (?,?,?,?,?,?,0,?,?,-1)";

  static String UPDATE_EVENT_TRACE_STATUS = "UPDATE event_trace SET status=?, last_update_date =? ,resp_code=?, resp_msg=? WHERE code =? AND object_key=? AND kind_id =?";

  static String UPDATE_EVENT_TRACE_STATUS_RETRY_TIMES = "UPDATE event_trace SET status=?, last_update_date =? ,resp_code=?, resp_msg=?, retry_times =? WHERE trace_id=?";


  static String QUERY_EVENT_TRACE = "SELECT trace_id,order_id,status,retry_times,resp_code,resp_msg FROM event_trace WHERE status = ? AND kind_id = ? AND code = ?";

  static String FIND_ALL_EVENT_BY_CODE =
      "SELECT et.object_key,et.code, em.order_type_id, et.status,ep.param_name,ep.param_value FROM event_trace et,event_meta em,event_params ep "
          + " WHERE et.object_key = em.object_key "
          + " AND em.meta_id = ep.meta_id "
          + " AND et.code = ? "
          + " AND et.kind_id = ? "
          + " AND et.status != ? "
          + " ORDER BY et.object_key";

  static String FIND_FAILED_EVENT =
      "SELECT trace_id,object_key,status,kind_id,retry_times FROM event_trace "
          + "WHERE object_key = (%s) "
          + " AND  status = ?"
          + " AND kind_id = ? "
          + " AND code = ？";

  @Autowired
  @Qualifier("dispatcherDataSource")
  private DataSource dataSource;


  /**
   * 根据接口编码查询该接口处理失败的信息
   *
   * @param code 接口编码
   * @param objectKeys 事件id集合
   * @param kindId 事件类型id
   * @return 如果存在失败订单，返回失败订单的信息，如果不存在，返回空集合
   */
  public List<EventTrace> findFailedEvent(String code, Set<String> objectKeys, Integer kindId) {
    Connection connection = null;
    PreparedStatement pstmt = null;
    List<EventTrace> eventTraces = new ArrayList<>();
    String sql = String.format(FIND_FAILED_EVENT, Joiner.on(",").join(objectKeys));
    try {
      connection = dataSource.getConnection();
      pstmt = connection.prepareStatement(sql);
      pstmt.setString(1, EventStatus.FAILED);
      pstmt.setInt(2, kindId);
      pstmt.setString(3, code);
      ResultSet rs = pstmt.executeQuery();
      while (rs.next()) {
        EventTrace eventTrace = new EventTrace();
        eventTrace.setCode(code);
        eventTrace.setTraceId(rs.getLong("trace_id"));
        eventTrace.setRetryTimes(rs.getInt("retry_times"));
        eventTrace.setStatus(rs.getString("status"));
        eventTrace.setObjectKey(rs.getString("object_key"));
        eventTrace.setEventType(rs.getInt("event_type"));
        eventTrace.setKindId(kindId);
        eventTraces.add(eventTrace);
      }
    } catch (Exception e) {
      log.error("find all event by code failed", e);
    } finally {
      JdbcUtils.close(pstmt);
      JdbcUtils.close(connection);
    }
    return eventTraces;
  }


  /**
   * 根据接口编码查找接口下所有未发送成功和发送失败的事件
   *
   * @param tasks 任务编号
   */
  public Map<String, Collection<EventMetaDataWrapper>> findAllEventByCode(List<String> tasks) {
    Connection connection = null;
    PreparedStatement pstmt = null;
    Map<String, Collection<EventMetaDataWrapper>> result = new HashMap<>();
    try {
      connection = dataSource.getConnection();
      pstmt = connection.prepareStatement(FIND_ALL_EVENT_BY_CODE);
      for (String task : tasks) {
        String[] codeKindId = task.split("-");
        String code = codeKindId[0];
        String kindId = codeKindId[1];
        Map<String, EventMetaDataWrapper> eventMetaDataMap = new HashMap<>();

        pstmt.setString(1, code);
        pstmt.setInt(2, Integer.valueOf(kindId));
        pstmt.setString(3, EventStatus.SUCCESS);
        ResultSet rs = pstmt.executeQuery();
        ListMultimap<String, Long> acctIds = ArrayListMultimap.create();
        ListMultimap<String, Long> offerIds = ArrayListMultimap.create();
        ListMultimap<String, Long> prodIds = ArrayListMultimap.create();
        ListMultimap<String, Long> custIds = ArrayListMultimap.create();
        ListMultimap<String, Long> offer2ProdIds = ArrayListMultimap.create();
        while (rs.next()) {
          EventMetaDataWrapper eventMetaDataWrapper = null;
          EventMetaData metaData = null;
          String objectKey = rs.getString("object_key");
          if (!eventMetaDataMap.containsKey(objectKey)) {
            eventMetaDataWrapper = new EventMetaDataWrapper();
            metaData = new EventMetaData();
            metaData.setBusinessNo(objectKey);
            metaData.setOrderTypeIds(rs.getString("order_type_id"));
            eventMetaDataWrapper.setEventMetaData(metaData);
            eventMetaDataWrapper.setStatus(rs.getString("status"));
            eventMetaDataMap.put(objectKey, eventMetaDataWrapper);
          }

          String name = rs.getString("param_name");
          String value = rs.getString("param_value");
          if (EventParamConst.PRODID.equals(name) && !"0".equals(value)) {
            prodIds.put(objectKey, Long.valueOf(value));
          } else if (EventParamConst.ACCTID.equals(name) && !"0".equals(value)) {
            acctIds.put(objectKey, Long.valueOf(value));

          } else if (EventParamConst.OFFER2PRODID.equals(name) && !"0".equals(value)) {
            offer2ProdIds.put(objectKey, Long.valueOf(value));
          } else if (EventParamConst.CUSTID.equals(name) && !"0".equals(value)) {
            custIds.put(objectKey, Long.valueOf(value));
          } else if (EventParamConst.OFFERID.equals(name) && !"0".equals(value)) {
            offerIds.put(objectKey, Long.valueOf(value));
          }
        }

        for (Entry<String, EventMetaDataWrapper> entry : eventMetaDataMap.entrySet()) {
          String objectKey = entry.getKey();
          EventMetaData metaData = entry.getValue().getEventMetaData();
          List<Long> acctIdList = acctIds.get(objectKey);
          List<Long> offerIdList = offerIds.get(objectKey);
          List<Long> prodIdList = prodIds.get(objectKey);
          List<Long> custIdList = custIds.get(objectKey);
          List<Long> offer2ProdIdList = offer2ProdIds.get(objectKey);
          metaData.setAcctId(acctIdList.toArray(new Long[acctIdList.size()]));
          metaData.setCustId(custIdList.toArray(new Long[acctIdList.size()]));
          metaData.setOfferProdId(offer2ProdIdList.toArray(new Long[offer2ProdIdList.size()]));
          metaData.setProdInstId(prodIdList.toArray(new Long[prodIdList.size()]));
          metaData.setOfferProdId(offerIdList.toArray(new Long[offerIdList.size()]));
        }

        result.put(code, eventMetaDataMap.values());
      }


    } catch (Exception e) {
      log.error("find all event by code failed", e);
    } finally {
      JdbcUtils.close(pstmt);
      JdbcUtils.close(connection);
    }

    return result;
  }

  /**
   * 根据接口编码、消费主题查询事件发送记录
   *
   * @param code 接口编码
   * @param kindId 事件种类id
   * @param status 事件处理状态
   * @return 所有事件集合
   */
  public List<EventTrace> queryEventTraceByCode(String code, Integer kindId, String status) {
    Connection connection = null;
    PreparedStatement pstmt = null;
    List<EventTrace> eventTraces = new ArrayList<>();
    try {
      connection = dataSource.getConnection();
      pstmt = connection.prepareStatement(QUERY_EVENT_TRACE);
      pstmt.setString(1, status);

      pstmt.setInt(2, kindId);
      pstmt.setString(3, code);

      ResultSet resultSet = pstmt.executeQuery();

      while (resultSet.next()) {
        EventTrace eventTrace = new EventTrace();
        eventTrace.setTraceId(resultSet.getLong("trace_id"));
        eventTrace.setRespCode(resultSet.getInt("resp_code"));
        eventTrace.setRetryTimes(resultSet.getInt("retry_times"));
        eventTrace.setStatus(resultSet.getString("status"));
        eventTrace.setObjectKey(resultSet.getString("object_key"));
        eventTrace.setRespMsg(resultSet.getString("resp_msg"));

        eventTrace.setCode(code);
        eventTrace.setKindId(kindId);

        eventTraces.add(eventTrace);
      }

    } catch (Exception e) {
      log.error("query event trace by code failed", e);
    } finally {
      JdbcUtils.close(pstmt);
      JdbcUtils.close(connection);
    }

    return eventTraces;
  }


  public int updateEventStatus(EventTrace eventTrace) {
    int result = 0;
    Connection connection = null;
    PreparedStatement pstmt = null;
    try {
      connection = dataSource.getConnection();
      pstmt = connection.prepareStatement(UPDATE_EVENT_TRACE_STATUS_RETRY_TIMES);
      pstmt.setString(1, eventTrace.getStatus());
      pstmt.setTimestamp(2, eventTrace.getLastUpdateDate());
      pstmt.setInt(3, eventTrace.getRespCode());
      if (eventTrace.getRespMsg().length() > 2000) {
        String subStr = eventTrace.getRespMsg().substring(0, 1990);
        eventTrace.setRespMsg(subStr);
      }
      pstmt.setString(4, eventTrace.getRespMsg());

      pstmt.setInt(5, eventTrace.getRetryTimes());
      pstmt.setLong(6, eventTrace.getTraceId());
      result = pstmt.executeUpdate();
      connection.commit();
    } catch (Exception e) {
      log.error("update event trace failed", e);
      try {
        connection.rollback();
      } catch (SQLException e1) {
        log.error("rollback error", e);
      }
    } finally {
      JdbcUtils.close(pstmt);
      JdbcUtils.close(connection);
    }
    return result;
  }

  public int updateEventStatus(String objectKey, Integer kindId, String code, String status,
      Timestamp lastUpdateDate, Integer respCode, String respMsg) {
    int result = 0;
    Connection connection = null;
    PreparedStatement pstmt = null;
    try {
      connection = dataSource.getConnection();
      pstmt = connection.prepareStatement(UPDATE_EVENT_TRACE_STATUS);
      pstmt.setString(1, status);
      pstmt.setTimestamp(2, lastUpdateDate);
      pstmt.setInt(3, respCode);
      if (respMsg.length() > 2000) {
        String newRespMsg = respMsg.substring(0, 1990);
        pstmt.setString(4, newRespMsg);
      } else {
        pstmt.setString(4, respMsg);
      }

      pstmt.setString(5, code);
      pstmt.setString(6, objectKey);
      pstmt.setInt(7, kindId);
      result = pstmt.executeUpdate();
      connection.commit();
    } catch (Exception e) {

      log.error("update event trace failed", e);
      try {
        connection.rollback();
      } catch (SQLException e1) {
        log.error("rollback error", e);
      }
    } finally {
      JdbcUtils.close(pstmt);
      JdbcUtils.close(connection);
    }
    return result;
  }

  public boolean batchUpdateEventStatus(List<SendResponse> responses, String status) {
    Connection connection = null;
    PreparedStatement pstmt = null;
    boolean result = false;
    try {
      connection = dataSource.getConnection();
      pstmt = connection.prepareStatement(UPDATE_EVENT_TRACE_STATUS);
      Timestamp timestamp = new Timestamp(System.currentTimeMillis());
      for (SendResponse response : responses) {
        Response resp = response.getResponse();
        pstmt.setString(1, status);
        pstmt.setTimestamp(2, timestamp);
        pstmt.setInt(3, resp.getCode());
        pstmt.setString(4, resp.getMsg());

        pstmt.setString(5, response.getCode());
        pstmt.setString(6, response.getEventId());
        pstmt.addBatch();
      }

      int counts[] = pstmt.executeBatch();
      result = counts.length > 0 ? true : false;
      connection.commit();
    } catch (Exception e) {

      log.error("update event trace failed", e);
      try {
        connection.rollback();
      } catch (SQLException e1) {
        log.error("rollback error", e);
      }
    } finally {
      JdbcUtils.close(pstmt);
      JdbcUtils.close(connection);
    }
    return result;
  }


  public int insertEventTrace(EventTrace eventTrace) {
    int result = 0;
    Connection connection = null;
    PreparedStatement pstmt = null;
    try {
      connection = dataSource.getConnection();
      pstmt = connection.prepareStatement(INSERT_EVENT_TRACE);
      pstmt.setLong(1, eventTrace.getTraceId());
      pstmt.setString(2, eventTrace.getObjectKey());
      pstmt.setString(3, eventTrace.getCode());
      pstmt.setString(4, eventTrace.getStatus());
      pstmt.setInt(5, eventTrace.getKindId());
      pstmt.setInt(6, eventTrace.getEventType());
      pstmt.setTimestamp(7, eventTrace.getCreateDate());
      pstmt.setTimestamp(8, eventTrace.getLastUpdateDate());

      result = pstmt.executeUpdate();
      connection.commit();

    } catch (Exception e) {
      log.error("insert event trace failed", e);
      try {
        connection.rollback();
      } catch (SQLException e1) {
        log.error("rollback error", e);
      }
    } finally {
      JdbcUtils.close(pstmt);
      JdbcUtils.close(connection);
    }

    return result;
  }


}
